package ruc.MongoDB.write;

import com.mongodb.client.MongoCollection;
import org.bson.Document;
import ruc.Consts.constant;
import ruc.utils.IOUtils;
import ruc.MongoDB.MongoDB_Common;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

public class MongoDB_ConcurrentWrite {
    private String IP, dbName, tbName;
    private int port;
    private ReentrantLock lock = new ReentrantLock();

    MongoDB_ConcurrentWrite(String IP, int port, String dbName, String tbName) {
        this.IP = IP;
        this.port = port;
        this.dbName = dbName;
        this.tbName = tbName;
    }

    long splitAndSend(String[] arr, int all, int num){
        lock.lock();
        MongoCollection<Document> conn = MongoDB_Common.GetMongoCollection(IP, port, dbName, tbName);
        List<Document> list = new ArrayList<>();
        try{
            for(int i = 0; i < arr.length; i++) {
                if (i % all != num) {
                    continue;
                }
                String[] val = arr[i].split(constant.Delimiter);
                Document document = new Document().append("ts",val[0]).append("userInfo", val[1]).append("server", val[2])
                                .append("hostname", val[3]).append("cpu_usage", val[4]).append("memory_usage", val[5]).append("disk_usage", val[6]);
                list.add(document);
            }
        }catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

        if(list.size() == 0) { // mongoDB does not support inserting into empty collections
            return 0;
        }
        long st_time = System.nanoTime();
        conn.insertMany(list);
        long ed_time = System.nanoTime();
        return (ed_time - st_time) / constant.Ns2Ms;
    }

    void MulInsertData(int[] ThreadNum, String[] val) throws InterruptedException {
        for(int i = 0; i < ThreadNum.length; i++) {
            CountDownLatch latch = new CountDownLatch(ThreadNum[i]);
            AtomicLong costTime = new AtomicLong(0);
            //clear data
            MongoCollection<Document> conn = MongoDB_Common.GetMongoCollection(IP, port, dbName, tbName);
            conn.drop();

            for(int k = 0; k < ThreadNum[i]; k++) {
                int finalJ = k;
                int finalI = i;
                new Thread(()->{
                    costTime.addAndGet(splitAndSend(val, ThreadNum[finalI], finalJ));
                    latch.countDown();
                }).start();
            }
            latch.await();
            System.out.printf("[TSBench-D] [MongoDB_ConcurrentWrite] >>>>>> threadNum %d, insert time %d ms >>>>>> \n", ThreadNum[i], costTime.longValue());
            IOUtils.WriteData(constant.GetWriteFilePath("MongoDB", "ConcurrentWrite"),
                    new String[]{String.valueOf(ThreadNum[i]), String.valueOf(costTime.longValue())});
        }
    }

    public static void main(String[] args) throws InterruptedException {
        MongoDB_ConcurrentWrite db = new MongoDB_ConcurrentWrite(constant.IP, constant.MongoDB_port, constant.dbName, constant.Concurrent_tbName);
        db.MulInsertData(constant.ConWrite_Thread, IOUtils.GetData());
    }
}
